package nsq

import (
	"context"
	"fmt"
	"github.com/nsqio/go-nsq"
	"github.com/pkg/errors"
	"go.uber.org/zap"
	"os"
	"pay/pkg/ecode"
	env2 "pay/pkg/env"
	"pay/pkg/log"
	"pay/pkg/request_key"
	"pay/pkg/wechat"
	"strings"
	"time"
)

var TopicContextKey = struct{}{}

type Consumer struct {
	*nsq.Consumer
}

type Handler struct {
	topic       string
	env         env2.Type
	processor   func(context.Context, []byte) error
	serviceName string
}

func (d Handler) HandleMessage(m *nsq.Message) (err error) {
	ctx := request_key.GenerateNSQContext()
	ctx = context.WithValue(ctx, TopicContextKey, d.topic)
	defer d.failOver(ctx, m.Body, time.Now(), d.env, &err)
	err = d.processor(ctx, m.Body)
	return
}

func (d *Handler) failOver(ctx context.Context, body []byte, startTime time.Time, env env2.Type, errAddr *error) {
	var errString, stackTrace, statusCode string
	logger := log.FromContext(ctx)
	if r := recover(); r != nil {
		errString = fmt.Sprintf("%s", r)
		stackTrace = ecode.GetPanicTrace("nsq", r)
		statusCode = "Fail-Unknown"
	} else if *errAddr != nil {
		switch v := errors.Cause(*errAddr).(type) {
		case *ecode.ErrorCode:
			errString = fmt.Sprintf("%s-%s", v.CodeMessage, v.ErrMessage)
			statusCode = fmt.Sprintf("Fail-%s", v.ErrorType)
		case ecode.WithStackModel:
			stackTrace = fmt.Sprintf("nsq error: %+v", *errAddr)
			errString = "系统错误"
			statusCode = "Fail-Unknown"
		default:
			// 错误未填充堆栈信息
			err := *errAddr
			stackTrace = fmt.Sprintf("nsq error: %s (no stack filled)", err.Error())
			errString = "系统错误"
			statusCode = "Fail-Unknown"
		}
		logger.Error(errString)
	} else {
		statusCode = "Success"
	}
	if stackTrace != "" {
		if env != env2.Local {
			wechat.AsyncSendMarkDown(
				ctx, wechat.FormatBugMessage(ctx, env, d.serviceName, d.topic, string(body), stackTrace))
		} else {
			_, _ = fmt.Fprintf(os.Stderr, stackTrace)
		}
	}
	if errString != "" || d.env != env2.Production {
		// 改为仅报错时打印日志
		logger.Infow("nsq",
			"status_code", statusCode,
			"latency", time.Since(startTime).Seconds(),
			"method", "consume",
			"path", d.topic,
			"body", string(body),
			"error", errString,
			"stacktrace", stackTrace,
		)
	}
}

type nsqLogger struct {
	logger *zap.SugaredLogger
}

// Output impl
func (n *nsqLogger) Output(_ int, s string) error {
	if strings.Contains(s, "TOPIC_NOT_FOUND") {
		// 无topic不需要报错
		return nil
	}
	n.logger.Warnw(s)
	return nil
}

func newConsumer(ctx context.Context, topic string, c *Configuration, env env2.Type,
	serviceName string, processor func(context.Context, []byte) error) (consumer *Consumer, err error) {
	consumer = &Consumer{}
	handler := &Handler{
		topic:       topic,
		env:         env,
		processor:   processor,
		serviceName: serviceName,
	}
	if consumer.Consumer, err = nsq.NewConsumer(handler.topic, c.Channel, c.Config); err != nil {
		err = fmt.Errorf("NSQ Consumer创建失败: topic=%s, error=%s", handler.topic, err)
		return
	}
	consumer.SetLogger(&nsqLogger{log.FromContext(ctx)}, nsq.LogLevelError)
	consumer.AddConcurrentHandlers(handler, c.ConsumeConcurrency)
	if err = consumer.ConnectToNSQLookupd(c.LookupAddress); err != nil {
		err = fmt.Errorf("NSQ Consumer链接失败, address=%s, error=%s", c.LookupAddress, err)
		return
	}
	return
}

// InitConsumers
func InitConsumers(ctx context.Context, c *Configuration, env env2.Type, serviceName string,
	processorMap map[string]func(context.Context, []byte) error) (consumers []*Consumer, err error) {
	log.FromContext(ctx).Infow("NSQ消费处理...加载中")
	// 订阅pos订单统计
	for topic, processor := range processorMap {
		var consumer *Consumer
		if consumer, err = newConsumer(ctx, topic, c, env, serviceName, processor); err != nil {
			return
		}
		consumers = append(consumers, consumer)
	}
	log.FromContext(ctx).Infow("NSQ消费处理...加载完成")
	return
}

func InitProducer(ctx context.Context, c *Configuration) (producer *nsq.Producer, err error) {
	producer, err = nsq.NewProducer(c.NSQAddress, c.Config)
	if err != nil {
		logger := log.FromContext(ctx)
		logger.Panic("initial nsq failed", zap.Error(err))
	}
	return
}
